package friot.service.cloud.config.mqtt;

import cn.hutool.core.lang.UUID;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.channel.ExecutorChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import javax.annotation.Resource;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author He Kuntao
 * @desc MQTT配置类DSL
 * @date 2024/2/11 13:18
 */
@Configuration
public class MqttConfigV2 {
    @Resource
    private MqttProperties mqttProperties;
    @Resource
    private MqttMessageHandler mqttMessageHandler;


    /**
     * MQTT 客户端工厂
     */
    @Bean
    public MqttPahoClientFactory mqttPahoClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(mqttProperties.getHost().split(","));
        options.setUserName(mqttProperties.getUsername());
        options.setPassword(mqttProperties.getPassword().toCharArray());
        options.setCleanSession(mqttProperties.isClearSession());
        factory.setConnectionOptions(options);
        return factory;
    }


    /**
     * MQTT 入站通道适配器
     *
     * @param factory MQTT工厂
     */
    @Bean
    public MqttPahoMessageDrivenChannelAdapter adapter(MqttPahoClientFactory factory) {
        return new MqttPahoMessageDrivenChannelAdapter(getClientId(), factory, mqttProperties.getDefaultTopic());
    }

    /**
     * MQTT 入站通道门面细节配置
     */
    @Bean
    public IntegrationFlow mqttInboundFlow(MqttPahoMessageDrivenChannelAdapter adapter) {
        return IntegrationFlows.from(adapter)
                .channel(new ExecutorChannel(mqttThreadPoolTaskExecutor()))
                .handle(mqttMessageHandler)
                .get();
    }


    /**
     * 获取客户端id
     *
     * @return 随机生产的客户端ID
     */
    private String getClientId() {
        return "frsf_" + UUID.fastUUID();
    }

    /**
     * Mqtt接受消息，处理消息的线程池
     * 可以让消息处理由同步变为异步
     */
    @Bean
    public ThreadPoolTaskExecutor mqttThreadPoolTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        // 最大可创建的线程数
        int maxPoolSize = 200;
        executor.setMaxPoolSize(maxPoolSize);
        // 核心线程池大小
        int corePoolSize = 50;
        executor.setCorePoolSize(corePoolSize);
        // 队列最大长度
        int queueCapacity = 1000;
        executor.setQueueCapacity(queueCapacity);
        // 线程池维护线程所允许的空闲时间
        int keepAliveSeconds = 300;
        executor.setKeepAliveSeconds(keepAliveSeconds);
        // 线程池对拒绝任务(无线程可用)的处理策略
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        return executor;
    }


    /**
     * MQTT 出站通道门面细节配置
     *
     * @param factory MQTT工厂
     */
    @Bean
    public IntegrationFlow mqttOutboundFlow(MqttPahoClientFactory factory) {
        // 出站通道适配器
        MqttPahoMessageHandler handler = new MqttPahoMessageHandler(getClientId(), factory);
        // 异步发送
        handler.setAsync(true);
        handler.setConverter(new DefaultPahoMessageConverter());
        handler.setDefaultTopic(mqttProperties.getDefaultTopic());
        return IntegrationFlows.from("mqttOutboundChannel")
                .handle(handler)
                .get();
    }


}
